import cannot_lowering
import unittest

class TestLowering(unittest.TestCase):
    # 在每个测试方法前执行
    def setUp(self):
        self.test_string = "TestLoweringTool"

    def test_lowering_registered(self):
        test_log = ["Skip lowering node test1, because No lowering registered for op Assign.\n"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', '算子类型未注册lowerng', 'Skip lowering node test1, because No lowering registered for op Assign.\n; log line: 1', '']])

    def test_lowering_skip_gather(self):
        test_log = ["Skip lowering node test1, as: Gather is in high_performance mode\n"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'Gather原算子为高性能模式', 'Skip lowering node test1, as: Gather is in high_performance mode\n; log line: 1', '']])

    def test_lowering_skip_node(self):
        test_log = ["Skip lowering node test1, as: Batch dims is not 0"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'lowering失败', 'Skip lowering node test1, as: Batch dims is not 0; log line: 1', '']])

    def test_lowering_skip_compatibility(self):
        test_log = ["Skip lowering node test1, because failed to check IR compatibility.\n"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', '算子原型不规范', 'Skip lowering node test1, because failed to check IR compatibility.\n; log line: 1', '']])

    def test_lowering_lift(self):
        test_log = ["Lift AscBackend node autofuse_pointwise_0_Add, node list is [Add(test1)], as: Num fused nodes 1 < 2\n"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(len(lifting_nodes), 1)
        self.assertEqual(len(lifting_original_nodes), 1)

    def test_lowering_lift_nodes(self):
        test_log = ["Lift AscBackend nodes [autofuse_pointwise_0_Add, autofuse_pointwise_1_Add], node list is [Add(test1)], as: Num fused nodes 1 < 2\n"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'lowering失败,待分析', 'Lift AscBackend nodes [autofuse_pointwise_0_Add, autofuse_pointwise_1_Add], node list is [Add(test1)], as: Num fused nodes 1 < 2\n; log line: 1', '']])
        self.assertEqual(len(lifting_nodes), 0)

    def test_lowering_dtypeunsupport(self):
        test_log = ["log\n", "InferDataType:Infer Op ops.Lt with input dtypes [DT_INT32, DT_INT32] got outputs\n","Fallback lowering for node test1, type add, as: kernel box has unsupported dtype."]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'ascirdtype不支持', 'log\nInferDataType:Infer Op ops.Lt with input dtypes [DT_INT32, DT_INT32] got outputs\nFallback lowering for node test1, type add, as: kernel box has unsupported dtype.; log line: 3', '']])

    def test_lowering_oversize(self):
        test_log = ["Fallback lowering for node test1, type add, as: Kernel box is oversize, unable to lowering after realize its inputs"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'Kernelbox超大', 'Fallback lowering for node test1, type add, as: Kernel box is oversize, unable to lowering after realize its inputs; log line: 1', '']])

    def test_lowering_fallback1(self):
        test_log = ["Fallback lowering for node test1, type add, as: This node has external kernel box"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'lowering失败', 'Fallback lowering for node test1, type add, as: This node has external kernel box; log line: 1', '']])

    def test_lowering_fallback2(self):
        test_log = ["Fallback lowering for node test1, type add, as: This node should not lowering, "
             "or not register lowering func, or unable to imply lowering"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', 'lowering失败', 'Fallback lowering for node test1, type add, as: This node should not lowering, '
             'or not register lowering func, or unable to imply lowering; log line: 1', '']])

    def test_lowering_fallback3(self):
        test_log = ["Fallback lowering for node cast, type add, as: This node has external kernel box"]
        nofuse_nodes = ["cast"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['cast', '', 'lowering', 'lowering失败', 'Fallback lowering for node cast, type add, as: This node has external kernel box; log line: 1', '']])

    def test_lowering_no_symbolic(self):
        test_log = ["Skip lowering node test1, as: No symbolic desc attr."]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(len(no_symbolize_nodes), 1)

    def test_lowering_no_sym_reduce(self):
        test_log = ["Skip lowering node test1, as: Input test1_input0 has no sym shape"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(len(no_symbolize_nodes), 1)

    def test_lowering_no_sym_reduce1(self):
        test_log = ["Skip lowering node test1, as: Output test1_output0 no sym shape"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(len(no_symbolize_nodes), 1)

    def test_lowering_no_find(self):
        test_log = ["Skip lowering node test1, as: Output test1_output0 no sym shape"]
        nofuse_nodes = ["test2"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test2', '', 'lowering', '日志未找到', '', '']])

    def test_lowering_fall_back1(self):
        test_log = ["Fall back lowering for node scope: Fused(Add(\"test1:0\"))[test1(Add)]. As Realize AscendC IR graph for kernel box Add(\"test1:0\")"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', '生成Ascbackend失败', 'Fall back lowering for node scope: Fused(Add(\"test1:0\"))[test1(Add)]. As Realize AscendC IR graph for kernel box Add(\"test1:0\"); log line: 1', '']])

    def test_lowering_fall_back2(self):
        test_log = ["Fall back lowering for node scope: Fused(Add(\"test1:0\"))[test1(Add)]. As unsupported scalar AscendC IR graph for kernel box Add(\"test1:0\")"]
        nofuse_nodes = ["test1"]
        outputs = []
        no_symbolize_nodes = []
        lifting_nodes = []
        lifting_original_nodes = []
        [no_symbolize_nodes, lifting_nodes, lifting_original_nodes] = cannot_lowering.cannot_lowering(test_log, nofuse_nodes, outputs)
        self.assertEqual(outputs, [['test1', '', 'lowering', '不支持纯scalar图', 'Fall back lowering for node scope: Fused(Add(\"test1:0\"))[test1(Add)]. As unsupported scalar AscendC IR graph for kernel box Add(\"test1:0\"); log line: 1', '']])
    # 在每个测试方法后执行
    def tearDown(self):
        pass

if __name__ == "__main__":
    unittest.main()